package com.zhao.demo.unbound.demo04_transformation.sample07_self.a_common

import com.zhao.demo.Raytek
import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * Description: 用户自定义函数之匿名方式(参数值动态传递)<br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/12/9 10:05
 * 需求:筛选出当天所有途经红外测温仪编号为raytek_2的旅客信息
 * 无界流中进行处理,kafka统一如此
 * 注意:和装载本地离线的文件内部处理机制不同
 * @author 柒柒
 * @version : 1.0
 */

object SelfDefineFunctionDemo2 {
  def main(args: Array[String]): Unit = {
    //1.前提:拦截非法的参数
    if (args == null || args.length != 6){
      sys.error(
        """
          |请输入参数! --hostname <主机名> --port <端口号> --id <红外测温仪的id>
          |""".stripMargin)
      sys.exit(-1)
    }

    //2.获得参数值
    val tool = ParameterTool.fromArgs(args)
    val hostname = tool.get("hostname")
    val port = tool.getInt("port")
    val id = tool.get("id")

    //3.环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //4.实时读取流数据,计算并显示结果
    import org.apache.flink.api.scala._

    env.socketTextStream(hostname,port)
      .map(perInfo =>{
        val arr: Array[String] = perInfo.split(",")
        val id = arr(0).trim
        val temperature = arr(1).trim.toDouble
        val name = arr(2).trim
        val timestamp = arr(3).trim.toLong
        val location = arr(4).trim
        Raytek(id, temperature, name, timestamp, location)
      }).filter(
      new FilterFunction[Raytek] {
        override def filter(value: Raytek): Boolean = {
          id.equals(value.id)
        }
      }).print("使用自定义<匿名>函数之动态传参,实现的效果->")

    //3.启动
    env.execute()
  }
}
